package com.example.rocket.controller;

import com.example.rocket.entity.UserEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Author ifredomvip@gmail.com
 * @Date 2022/8/16 16:21
 * @Version 1.0.0
 * @Description
 **/
@Slf4j
@RestController
@RequestMapping("/sendMessage")
public class ProducerMessage {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    @Value(value = "topic-string")
    private String stringTopic;

    @Value(value = "topic-user")
    private String userTopic;

    @Value(value = "topic-tag")
    private String tagTopic;

    /**
     * 发送异步消息
     */
    @GetMapping("/async")
    public void sendAsyncMessage() {

        String msg = "消息内容";

        rocketMQTemplate.asyncSend(stringTopic, MessageBuilder.withPayload(msg).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("消息发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("消息发送失败", throwable);
            }
        });
        log.info("=======发送消息完成=========");
    }

    /**
     * 发送同步消息
     */
    @GetMapping
    public void sendSyncMessage() {

        // 发送对象类型消息  User
        UserEntity user = new UserEntity();
        user.setUsername("ifredom");
        user.setAge(18);

        rocketMQTemplate.convertAndSend(userTopic, user);

        log.info("=======发送消息完成=========");
    }

    /**
     * 发送事务信息
     */
    @GetMapping("/transaction")
    public void sendTransactionMessage() throws InterruptedException {

        String msg = "事务消息内容";
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            Message<String> message = MessageBuilder.withPayload(msg).build();
            String destination = tagTopic + ":" + tags[i % tags.length];
            SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, destination);
            log.info(String.valueOf(sendResult));
            Thread.sleep(10);
        }

        log.info("=======发送消息完成=========");

    }
}
